package tmall.framework.kafka;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import tmall.framework.anotation.Topic;

import javax.annotation.Resource;

@Component
@Slf4j
public abstract class KafkaConsumer<K,V> {
    @Resource
    @Transactional(rollbackFor = Exception.class)
    @KafkaListener
    protected void consumerMessage(ConsumerRecord<K,V> record, Consumer consumer){
        Topic topic =  this.getClass().getAnnotation(Topic.class);
        String name = topic.name();
        if(StringUtils.isEmpty(name)){
            return;
        }
        if(record == null||record.value() == null){
            log.error("consumer value is not exist");
        }
        consumer(record.value());

    }

    abstract public void consumer(V v);

    private void offsetCommit(){}


}
